package com.test.textGame.mq;

import com.alibaba.fastjson.JSONObject;
import com.test.textGame.algorithm.TopK;
import com.test.textGame.service.Top10Service;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@Slf4j
public class RocketMqSlaverListener implements MessageListenerConcurrently {

    @Autowired
    private Top10Service top10Service;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        log.info("收到slaver的消息");
        try{
            for (MessageExt message : list) {
                String data = new String(message.getBody());
                TopK.Node node = JSONObject.parseObject(data, TopK.Node.class);
                top10Service.consumeSlaverMsg(node.player, node.score);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }catch (Exception e){
            log.error("处理消费者数据发生异常。{}", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}